package us.luosl.hadoop.mr.io;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.MapWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.util.LineReader;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 自定义输入输出
 * Created by luosl on 2018/9/3.
 */
public class JsonInputFormat extends FileInputFormat{

    @Override
    public RecordReader createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
        return new JsonRecordReader();
    }

    public static class JsonRecordReader extends RecordReader<LongWritable, MapWritable>{
        private FileSplit fileSplit;
        private LineReader reader;
        private Text text = new Text();
        private LongWritable key = new LongWritable();
        private MapWritable value = new MapWritable();
        private AtomicLong count = new AtomicLong(1);

        @Override
        public void initialize(InputSplit split, TaskAttemptContext context) throws IOException, InterruptedException {
            fileSplit = (FileSplit) split;
            Path path = fileSplit.getPath();
            FileSystem fs = path.getFileSystem(context.getConfiguration());
            reader = new LineReader(fs.open(path));
        }

        /**
         * 这个方法会被调用多次   这个方法的返回值如果是 true 就会被调用一次
         * 每当nextKeyValue被调用一次 ，getCurrentKey，getCurrentValue 也会被跟着调用一次
         * getCurrentKey,getCurrentValue 给 Mapper 传key,value
         * @return
         * @throws IOException
         * @throws InterruptedException
         */
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            if(reader.readLine(text)!=0){
                JSONObject json = JSON.parseObject(text.toString());
                key.set(count.getAndIncrement());
                for(String key:json.keySet()){
                    value.put(new Text(key), new Text(json.get(key).toString()));
                }
                return true;
            }else{
                return false;
            }

        }

        @Override
        public LongWritable getCurrentKey() throws IOException, InterruptedException {
            return key;
        }

        @Override
        public MapWritable getCurrentValue() throws IOException, InterruptedException {
            return value;
        }

        @Override
        public float getProgress() throws IOException, InterruptedException {
            return 0;
        }

        @Override
        public void close() throws IOException {
            IOUtils.closeStream(reader);
        }
    }
}
